package dsx.lockProxy;

import dsx.config.Config;
import dsx.console.DsxConsole;
import dsx.log.ServLog;
import dsx.macro.ReturnValue;
import dsx.rpcProxy.RPCProxy;
import dsx.system.IThreadExWork;
import dsx.system.ThreadEx;
import dsx.system.ThreadExPool;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;

public class Lock {

    // RPC指令
    private static int lockCmd = 100000;
    // 服务列表
    private static ArrayList<String> serverList = new ArrayList<String>();
    // 线程池
    private static ThreadExPool threadPool = new ThreadExPool();
    private static int threadStackSize = 65536;
    private static int maxWorkerNum = 100;
    // 锁业务执行守护
    private static LockGuard lockGuard = new LockGuard();
    private static Queue<AsyncLock> asyncLockQueue = new LinkedList<AsyncLock>();
    // 解锁业务执行守护
    private static ReleaseGuard releaseGuard = new ReleaseGuard();
    private static Queue<AsyncRelease> asyncReleaseQueue = new LinkedList<AsyncRelease>();
    // 投票半数
    private static int voteHalf = 0;
    // 初始化状态标识
    private static boolean initialized = false;

    /// 检查初始化
    private static synchronized boolean checkConfig() {
        if (initialized)
            return true;

        // RPC指令
        try {
            String tmp = Config.get("Lock", "lockCmd");
            if (tmp != null && !tmp.isEmpty()) {
                int cmd = Integer.parseInt(tmp);
                lockCmd = cmd;
            }
        } catch (Exception ex) {
            DsxConsole.out(ex.getMessage(), "lock", ServLog.info);
            return false;
        }

        // 线程池新建线程的线程栈大小
        try {
            String tmp = Config.get("Lock", "threadStackSize");
            if (tmp != null && !tmp.isEmpty()) {
                int size = Integer.parseInt(tmp);
                threadStackSize = size;
            }
            threadPool.setThreadStackSize(threadStackSize);
        } catch (Exception ex) {
            DsxConsole.out(ex.getMessage(), "lock", ServLog.info);
            return false;
        }

        // 线程池最大工作线程数量
        try {
            String tmp = Config.get("Lock", "maxWorkerNum");
            if (tmp != null && !tmp.isEmpty()) {
                int workerNum = Integer.parseInt(tmp);
                maxWorkerNum = workerNum;
            }
            threadPool.setMaxWorkerNum(maxWorkerNum);
        } catch (Exception ex) {
            DsxConsole.out(ex.getMessage(), "lock", ServLog.info);
            return false;
        }

        // 锁服务器列表
        try {
            serverList.clear();
            String tmp = Config.get("Lock", "proxy");
            if (tmp != null && !tmp.isEmpty()) {
                String[] tmpLst = tmp.split(",");
                for (int i = 0; i < tmpLst.length; i++)
                    serverList.add(tmpLst[i].trim());
            }
            if (serverList.isEmpty())
                serverList.add("lock");
        } catch (Exception ex) {
            DsxConsole.out(ex.getMessage(), "lock", ServLog.info);
            return false;
        }

        // 投票半数，超过此数量投票通过
        voteHalf = serverList.size() / 2;

        // 线程池添加工作类型
        threadPool.addWork(LockGuard.guardType, lockGuard);
        threadPool.addWork(ReleaseGuard.guardType, releaseGuard);

        initialized = true;
        return true;
    }

    /// 锁业务执行
    private static boolean lock(LockReq req, int waitTimeout) {
        if (serverList.isEmpty())
            return false;
        if (serverList.size() == 1) {
            AsyncLock asyncLock = new AsyncLock(serverList.get(0), req, waitTimeout, 0);
            doLock(asyncLock);
            return (asyncLock.status == 2);
        }

        ArrayList<AsyncLock> asyncLockList = new ArrayList<AsyncLock>();
        for (int i = 0; i < serverList.size(); i++) {
            synchronized (asyncLockQueue) {
                AsyncLock asyncLock = new AsyncLock(serverList.get(i), req, waitTimeout, 0);
                asyncLockQueue.add(asyncLock);
                asyncLockList.add(asyncLock);
            }
            threadPool.notify(LockGuard.guardType);
        }
        ThreadEx.sleepEx(200);

        boolean result = false;
        int waitTime = 0;
        do {
            int sucNum = 0;
            int failNum = 0;
            for (int i = 0; i < asyncLockList.size(); i++) {
                switch (asyncLockList.get(i).status) {
                    case 2:
                        sucNum++;
                        break;
                    case 3:
                        failNum++;
                        break;
                    default:
                        break;
                }
            }
            if (sucNum > voteHalf) {
                result = true;
                break;
            }
            if (failNum > voteHalf)
                break;
            ThreadEx.sleepEx(200);
            waitTime += 200;
        } while (waitTime < waitTimeout);

        synchronized (asyncLockQueue) {
            for (int i = 0; i < asyncLockList.size(); i++) {
                switch (asyncLockList.get(i).status) {
                    case 0:
                    case 1:
                        asyncLockList.get(i).status = 9;
                        break;
                    case 2:
                        if (!result) {
                            req.type += 2;
                            doRelease(new AsyncRelease(serverList.get(i), req));
                        }
                        break;
                    default:
                        break;
                }
            }
        }

        return result;
    }

    /// 指向具体单服务器的锁业务执行
    private static void doLock(AsyncLock asyncLock) {
        int waitTime = 0;
        if (asyncLock.status == 0)
            asyncLock.status = 1;
        do {
            if (asyncLock.status == 9)
                return;
            Object answ = RPCProxy.request(asyncLock.lockServerName, asyncLock.req, LockAnsw.class);
            if (answ != null
                    && ((LockAnsw) answ).getRet() == ReturnValue.succeed) {
                asyncLock.status = 2;
                return;
            }
            ThreadEx.sleepEx(200);
            waitTime += 200;
        } while (waitTime < asyncLock.waitTimeout);
        asyncLock.status = 3;
    }

    /// 解锁业务执行
    private static void release(LockReq req) {
        if (serverList.isEmpty())
            return;
        if (serverList.size() == 1) {
            synchronized (asyncReleaseQueue) {
                asyncReleaseQueue.offer(new AsyncRelease(serverList.get(0), req));
            }
            threadPool.notify(ReleaseGuard.guardType);
            return;
        }
        for (int i = 0; i < serverList.size(); i++) {
            synchronized (asyncReleaseQueue) {
                asyncReleaseQueue.offer(new AsyncRelease(serverList.get(i), req));
            }
            threadPool.notify(ReleaseGuard.guardType);
        }
    }

    /// 指向具体单服务器的解锁业务执行
    private static void doRelease(AsyncRelease asyncRelease) {
        RPCProxy.request(asyncRelease.lockServerName, asyncRelease.req, LockAnsw.class);
    }

    static class AsyncLock {
        String lockServerName;
        LockReq req;
        int waitTimeout;
        int status; // 0 待执行，1 正在执行，2 已成功，3 已失败，9 已取消

        public AsyncLock(String lockServerName, LockReq req, int waitTimeout, int status) {
            this.lockServerName = lockServerName;
            this.req = req;
            this.waitTimeout = waitTimeout;
            this.status = status;
        }

    }

    static class LockGuard implements IThreadExWork {

        static final String guardType = "lockGuard";

        @Override
        public void run() {
            while (true) {
                try {
                    AsyncLock asyncLock;
                    synchronized (asyncLockQueue) {
                        asyncLock = asyncLockQueue.poll();
                    }
                    if (asyncLock != null)
                        doLock(asyncLock);
                    else
                        break;
                } catch (Exception ex) {
                    break;
                }
            }
        }
    }

    static class AsyncRelease {
        String lockServerName;
        LockReq req;

        public AsyncRelease(String lockServerName, LockReq req) {
            this.lockServerName = lockServerName;
            this.req = req;
        }

    }

    static class ReleaseGuard implements IThreadExWork {

        static final String guardType = "releaseGuard";

        @Override
        public void run() {
            while (true) {
                try {
                    AsyncRelease asyncRelease;
                    synchronized (asyncReleaseQueue) {
                        asyncRelease = asyncReleaseQueue.poll();
                    }
                    if (asyncRelease != null)
                        doRelease(asyncRelease);
                    else
                        break;
                } catch (Exception ex) {
                    break;
                }
            }
        }
    }

    public static boolean lockRow(String collectionKey, String rowKey,
                                  String ownerKey, int lockTimeout, int waitTimeout) {
        if (!checkConfig())
            return false;
        LockReq req = new LockReq();
        req.setCmd(lockCmd);
        req.type = 1;
        req.collectionKeys = new ArrayList<String>();
        req.collectionKeys.add(collectionKey);
        req.rowKeys = new ArrayList<String>();
        req.rowKeys.add(rowKey);
        req.ownerKey = ownerKey;
        req.lockTimeout = lockTimeout;

        return lock(req, waitTimeout);
    }

    public static boolean lockRows(String collectionKey, List<String> rowKeys,
                                   String ownerKey, int lockTimeout, int waitTimeout) {
        if (!checkConfig())
            return false;
        LockReq req = new LockReq();
        req.setCmd(lockCmd);
        req.type = 1;
        req.collectionKeys = new ArrayList<String>();
        for (int i = 0; i < rowKeys.size(); i++)
            req.collectionKeys.add(collectionKey);
        req.rowKeys = rowKeys;
        req.ownerKey = ownerKey;
        req.lockTimeout = lockTimeout;

        return lock(req, waitTimeout);
    }

    public static boolean lockRows(List<String> collectionKeys,
                                   List<String> rowKeys, String ownerKey, int lockTimeout,
                                   int waitTimeout) {
        if (!checkConfig())
            return false;
        LockReq req = new LockReq();
        req.setCmd(lockCmd);
        req.type = 1;
        req.collectionKeys = collectionKeys;
        req.rowKeys = rowKeys;
        req.ownerKey = ownerKey;
        req.lockTimeout = lockTimeout;

        return lock(req, waitTimeout);
    }

    public static boolean lockCollection(String collectionKey, String ownerKey,
                                         int lockTimeout, int waitTimeout) {
        if (!checkConfig())
            return false;
        LockReq req = new LockReq();
        req.setCmd(lockCmd);
        req.type = 0;
        req.collectionKeys = new ArrayList<String>();
        req.collectionKeys.add(collectionKey);
        req.rowKeys = null;
        req.ownerKey = ownerKey;
        req.lockTimeout = lockTimeout;

        return lock(req, waitTimeout);
    }

    public static boolean lockCollections(List<String> collectionKeys,
                                          String ownerKey, int lockTimeout, int waitTimeout) {
        if (!checkConfig())
            return false;
        LockReq req = new LockReq();
        req.setCmd(lockCmd);
        req.type = 0;
        req.collectionKeys = collectionKeys;
        req.rowKeys = null;
        req.ownerKey = ownerKey;
        req.lockTimeout = lockTimeout;

        return lock(req, waitTimeout);
    }

    public static void releaseRow(String collectionKey, String rowKey,
                                  String ownerKey) {
        if (!checkConfig())
            return;
        LockReq req = new LockReq();
        req.setCmd(lockCmd);
        req.type = 3;
        req.collectionKeys = new ArrayList<String>();
        req.collectionKeys.add(collectionKey);
        req.rowKeys = new ArrayList<String>();
        req.rowKeys.add(rowKey);
        req.ownerKey = ownerKey;
        req.lockTimeout = 0;

        release(req);
    }

    public static void releaseRows(String collectionKey, List<String> rowKeys,
                                   String ownerKey) {
        if (!checkConfig())
            return;
        LockReq req = new LockReq();
        req.setCmd(lockCmd);
        req.type = 3;
        req.collectionKeys = new ArrayList<String>();
        for (int i = 0; i < rowKeys.size(); i++)
            req.collectionKeys.add(collectionKey);
        req.rowKeys = rowKeys;
        req.ownerKey = ownerKey;
        req.lockTimeout = 0;

        release(req);
    }

    public static void releaseRows(List<String> collectionKeys,
                                   List<String> rowKeys, String ownerKey) {
        if (!checkConfig())
            return;
        LockReq req = new LockReq();
        req.setCmd(lockCmd);
        req.type = 3;
        req.collectionKeys = collectionKeys;
        req.rowKeys = rowKeys;
        req.ownerKey = ownerKey;
        req.lockTimeout = 0;

        release(req);
    }

    public static void releaseCollection(String collectionKey, String ownerKey) {
        if (!checkConfig())
            return;
        LockReq req = new LockReq();
        req.setCmd(lockCmd);
        req.type = 2;
        req.collectionKeys = new ArrayList<String>();
        req.collectionKeys.add(collectionKey);
        req.rowKeys = null;
        req.ownerKey = ownerKey;
        req.lockTimeout = 0;

        release(req);
    }

    public static void releaseCollections(List<String> collectionKeys,
                                          String ownerKey) {
        if (!checkConfig())
            return;
        LockReq req = new LockReq();
        req.setCmd(lockCmd);
        req.type = 2;
        req.collectionKeys = collectionKeys;
        req.rowKeys = null;
        req.ownerKey = ownerKey;
        req.lockTimeout = 0;

        release(req);
    }

}
